package com.pine.dcts.task;

import com.alibaba.datax.common.constant.PluginType;
import com.alibaba.datax.common.util.Configuration;
import com.pine.dcts.DctsEngine;
import com.pine.dcts.core.IPluginConfigContext;
import com.pine.dcts.core.impl.CoreConfigHandler;
import com.pine.dcts.service.TaskRunnerCallback;

/**
 * @author xiaoyuan
 * @create 2021/4/9 16:42
 **/
public class DataxTaskRunner extends TaskRunner<DataxTask> {

    public DataxTaskRunner(DataxTask task, TaskRunnerCallback callback) {
        super(task, callback);
        this.task = task;
    }

    @Override
    public Boolean execute() {
        Configuration configuration = Configuration.from(this.task.getTaskConfig());
        new CoreConfigHandler().supplementCoreConfig(configuration);
        task.getJobVO().getItems().forEach(item -> IPluginConfigContext.getPluginConfigHandlers().stream()
                .filter(h -> h.support(item.getName())).findFirst().ifPresent(pluginConfigHandler ->
                        pluginConfigHandler.supplementConfig(PluginType.valueOf(item.getType().toUpperCase())
                        , configuration)));
        new DctsEngine().start(configuration.toString());
        return true;
    }


}
